package com.ververica.cdc.guass.sink.jdbc.core.table.sink;

import com.ververica.cdc.guass.sink.jdbc.JdbcExecutionOptions;
import com.ververica.cdc.guass.sink.jdbc.core.database.dialect.JdbcDialect;
import com.ververica.cdc.guass.sink.jdbc.core.database.dialect.JdbcDialectConverter;
import com.ververica.cdc.guass.sink.jdbc.datasource.connections.SimpleJdbcConnectionProvider;
import com.ververica.cdc.guass.sink.jdbc.internal.JdbcOutputFormat;
import com.ververica.cdc.guass.sink.jdbc.internal.executor.*;
import com.ververica.cdc.guass.sink.jdbc.internal.options.InternalJdbcConnectionOptions;
import com.ververica.cdc.guass.sink.jdbc.internal.options.JdbcDmlOptions;
import com.ververica.cdc.guass.sink.jdbc.util.OperationType;
import com.ververica.cdc.guass.source.kafka.data.NullableValue;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;

import java.io.Serializable;
import java.util.Arrays;
import java.util.function.Function;

import static org.apache.flink.table.data.RowData.createFieldGetter;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;

/**
 * Builder for {@link JdbcOutputFormat} for Table/SQL.
 */
public class JdbcOutputFormatBuilder implements Serializable {

    private static final long serialVersionUID = 1L;

    private InternalJdbcConnectionOptions jdbcOptions;
    private JdbcExecutionOptions executionOptions;
    private JdbcDmlOptions dmlOptions;
    private DataType[] fieldDataTypes;

    public JdbcOutputFormatBuilder() {
    }

    public JdbcOutputFormatBuilder setJdbcOptions(InternalJdbcConnectionOptions jdbcOptions) {
        this.jdbcOptions = jdbcOptions;
        return this;
    }

    public JdbcOutputFormatBuilder setJdbcExecutionOptions(JdbcExecutionOptions executionOptions) {
        this.executionOptions = executionOptions;
        return this;
    }

    public JdbcOutputFormatBuilder setJdbcDmlOptions(JdbcDmlOptions dmlOptions) {
        this.dmlOptions = dmlOptions;
        return this;
    }

    public JdbcOutputFormatBuilder setFieldDataTypes(DataType[] fieldDataTypes) {
        this.fieldDataTypes = fieldDataTypes;
        return this;
    }

    /**
     * 检查所有必需的配置选项是否已设置。
     * 将字段数据类型转换为逻辑类型。
     * 根据 DML 选项中是否存在键字段，选择创建相应的执行器（Executor）：
     * 如果存在键字段，则创建 BufferReduceExecutor，用于处理 Upsert 查询（更新或插入操作）。
     * 如果不存在键字段，直接抛出异常
     *
     * @return
     */
    public JdbcOutputFormat<RowData, ?, ?> build() {
        checkNotNull(jdbcOptions, "jdbc options can not be null");
        checkNotNull(dmlOptions, "jdbc dml options can not be null");
        checkNotNull(executionOptions, "jdbc execution options can not be null");

        final LogicalType[] logicalTypes =
                Arrays.stream(fieldDataTypes)
                        .map(DataType::getLogicalType)
                        .toArray(LogicalType[]::new);

        if (dmlOptions.getKeyFields().isPresent() && dmlOptions.getKeyFields().get().length > 0) {
            // upsert query
            return new JdbcOutputFormat<>(
                    new SimpleJdbcConnectionProvider(jdbcOptions),
                    executionOptions,
                    () -> createBufferReduceExecutor(dmlOptions, logicalTypes));
        } else {
            // append only query
            return new JdbcOutputFormat<>(
                    new SimpleJdbcConnectionProvider(jdbcOptions),
                    executionOptions,
                    () ->
                            createSimpleBufferedExecutor(
                                    dmlOptions.getDialect(),
                                    dmlOptions.getFieldNames(),
                                    null,
                                    logicalTypes,
                                    dmlOptions.getTableName(),
                                    OperationType.Insert));

        }
    }

    private static JdbcBatchStatementExecutor<RowData> createBufferReduceExecutor(
            JdbcDmlOptions opt, LogicalType[] fieldTypes) {
        checkArgument(opt.getKeyFields().isPresent());
        JdbcDialect dialect = opt.getDialect();
        String tableName = opt.getTableName();
        String[] pkNames = opt.getKeyFields().get();
        int[] pkFields =
                Arrays.stream(pkNames)
                        .mapToInt(Arrays.asList(opt.getFieldNames())::indexOf)
                        .toArray();
        LogicalType[] pkTypes =
                Arrays.stream(pkFields).mapToObj(f -> fieldTypes[f]).toArray(LogicalType[]::new);

        return new TableBufferReducedStatementExecutor(
                createUpsertRowExecutor(
                        dialect,
                        tableName,
                        opt.getFieldNames(),
                        fieldTypes,
                        pkFields,
                        pkNames,
                        pkTypes),
                createDeleteExecutor(dialect, tableName, pkNames, pkTypes),
                createRowKeyExtractor(fieldTypes, pkFields),
                fieldTypes,
                opt.getFieldNames());
    }

    private static JdbcBatchStatementExecutor<RowData> createSimpleBufferedExecutor(
            JdbcDialect dialect, String[] fieldNames, String[] pkNames, LogicalType[] fieldTypes, String tableName, OperationType operationType) {

        return new TableBufferedStatementExecutor(
                createSimpleRowExecutor(dialect, fieldNames, pkNames, fieldTypes, tableName, operationType));
    }

    /**
     * createUpsertRowExecutor 方法根据 JDBC 方言的 Upsert 语句支持情况，选择创建相应的执行器。
     * 如果 JDBC 方言支持 Upsert 语句，就直接创建简单的行执行器；
     * 如果不支持，则创建 Insert 或 Update 执行器。这样可以兼容不同数据库的 Upsert 操作，提供更好的灵活性和适配性。
     */
    private static JdbcBatchStatementExecutor<RowData> createUpsertRowExecutor(
            JdbcDialect dialect,
            String tableName,
            String[] fieldNames,
            LogicalType[] fieldTypes,
            int[] pkFields,
            String[] pkNames,
            LogicalType[] pkTypes) {

        return dialect.supportUpsert() ? createSimpleRowExecutor(dialect, fieldNames, pkNames, fieldTypes, tableName, OperationType.Upsert)
                : createInsertOrUpdateExecutor(
                dialect,
                tableName,
                fieldNames,
                fieldTypes,
                pkFields,
                pkNames,
                pkTypes);
    }

    private static JdbcBatchStatementExecutor<RowData> createDeleteExecutor(
            JdbcDialect dialect, String tableName, String[] pkNames, LogicalType[] pkTypes) {

        return createSimpleRowExecutor(dialect, null, pkNames, pkTypes, tableName, OperationType.Delete);
    }


    private static JdbcBatchStatementExecutor<RowData> createSimpleRowExecutor(
            JdbcDialect dialect, String[] fieldNames, String[] pkNames, LogicalType[] fieldTypes, String tableName, OperationType operationType) {
        final JdbcDialectConverter rowConverter = dialect.getRowConverter(RowType.of(fieldTypes));
        return new TableSimpleDynamicFieldsStatementExecutor(dialect, rowConverter, fieldNames, pkNames, fieldTypes, tableName, operationType);
    }

    private static JdbcBatchStatementExecutor<RowData> createInsertOrUpdateExecutor(
            JdbcDialect dialect,
            String tableName,
            String[] fieldNames,
            LogicalType[] fieldTypes,
            int[] pkFields,
            String[] pkNames,
            LogicalType[] pkTypes) {

        return new TableInsertOrUpdateDynamicFieldsStatementExecutor(
                dialect,
                dialect.getRowConverter(RowType.of(pkTypes)),
                tableName,
                fieldNames,
                fieldTypes,
                pkNames,
                createRowKeyExtractor(fieldTypes, pkFields));
    }

    /**
     * 组装带有主键的RowKey
     *
     * @param logicalTypes
     * @param pkFields
     * @return
     */
    private static Function<RowData, RowData> createRowKeyExtractor(
            LogicalType[] logicalTypes, int[] pkFields) {
        final RowData.FieldGetter[] fieldGetters = new RowData.FieldGetter[pkFields.length];
        for (int i = 0; i < pkFields.length; i++) {
            fieldGetters[i] = createFieldGetter(logicalTypes[pkFields[i]], pkFields[i]);
        }

        // TableInsertOrUpdateDynamicFieldsStatementExecutor中执行
        // RowData pk = keyExtractor.apply(convertRowData);时会调用getPrimaryKey(row, fieldGetters)
        // 返回一个 Function，该 Function 提取主键
        return row -> getPrimaryKey(row, fieldGetters);
    }



    private static RowData getPrimaryKey(RowData row, RowData.FieldGetter[] fieldGetters) {
        GenericRowData pkRow = new GenericRowData(fieldGetters.length);

        RowData newRowData = new GenericRowData(row.getArity());
        for (int i = 0; i < row.getArity(); i++) {
            Object field = ((GenericRowData) row).getField(i);
            if(field instanceof NullableValue){
                ((GenericRowData) newRowData).setField(i, ((NullableValue<?>) field).getValue());
            }else {
                ((GenericRowData) newRowData).setField(i, field);
            }
        }

        for (int i = 0; i < fieldGetters.length; i++) {

            pkRow.setField(i, fieldGetters[i].getFieldOrNull(newRowData));

        }
        return pkRow;
    }
}
